package org.jetlinks.pro.cassandra.metadata;

import lombok.Setter;
import org.apache.commons.collections.CollectionUtils;
import org.bouncycastle.util.encoders.Base64;
import org.hswebframework.ezorm.core.param.Term;
import org.hswebframework.ezorm.rdb.executor.SqlRequest;
import org.hswebframework.ezorm.rdb.metadata.RDBSchemaMetadata;
import org.hswebframework.ezorm.rdb.metadata.RDBTableMetadata;
import org.hswebframework.ezorm.rdb.metadata.TableOrViewMetadata;
import org.hswebframework.ezorm.rdb.operator.DatabaseOperator;
import org.hswebframework.ezorm.rdb.operator.builder.DefaultQuerySqlBuilder;
import org.hswebframework.ezorm.rdb.operator.builder.FragmentBlock;
import org.hswebframework.ezorm.rdb.operator.builder.fragments.BlockSqlFragments;
import org.hswebframework.ezorm.rdb.operator.builder.fragments.PrepareSqlFragments;
import org.hswebframework.ezorm.rdb.operator.builder.fragments.SqlFragments;
import org.hswebframework.ezorm.rdb.operator.dml.query.QueryOperatorParameter;
import org.hswebframework.ezorm.rdb.operator.dml.query.SortOrder;
import org.jetlinks.pro.utils.ScrollPagingUtils;
import org.jetlinks.reactor.ql.utils.CastUtils;
import org.springframework.util.StringUtils;

import java.nio.ByteBuffer;
import java.util.*;

/**
 * Cassandra 查询SQL构造器,用于使用Cassandra进行查询时构造SQL。
 * <p>
 * Cassandra 对查询支持较弱,查询会有很多限制: 比如只能使用主键和索引列作为查询条件和排序条件.
 * <p>
 * 默认查询不会限制分页深度,在单个分区数据量较大时,建议设置:
 *
 * <pre>{@code
 * java -Dcassandra.query.max-page-index=1024
 * }</pre>
 *
 * @author zhouhao
 * @see DatabaseOperator#dml()
 * @see org.hswebframework.ezorm.rdb.operator.DMLOperator#query(String)
 * @see org.hswebframework.web.api.crud.entity.QueryParamEntity
 * @since 1.10
 */
public class CassandraQuerySqlBuilder extends DefaultQuerySqlBuilder {

    private static final int QUERY_MAX_PAGE_INDEX = Integer.getInteger("cassandra.query.max-page-index", -1);

    protected CassandraQuerySqlBuilder(RDBSchemaMetadata schema) {
        super(schema);
    }

    @Setter
    private int maxPageIndex = QUERY_MAX_PAGE_INDEX;

    @Override
    protected SqlFragments from(TableOrViewMetadata metadata, QueryOperatorParameter parameter) {
        //from {table}
        return PrepareSqlFragments
            .of()
            .addSql("from")
            .addSql(metadata.getFullName());
    }

    @Override
    protected SqlRequest build(TableOrViewMetadata metadata, QueryOperatorParameter parameter) {
        BlockSqlFragments fragments = BlockSqlFragments.of();

        Set<String> columns = new HashSet<>();

        for (Term term : parameter.getWhere()) {
            String key = term.getColumn() + term.getTermType();
            //对重复的查询列进行处理,不再拼接查询
            if (columns.contains(key)) {
                term.setValue(null);
            }
            if (term.getValue() != null) {
                columns.add(key);
            }
        }
        //select
        fragments.addBlock(FragmentBlock.before, "select");

        //columns
        fragments.addBlock(FragmentBlock.selectColumn, select(parameter, metadata)
            .orElseGet(() -> PrepareSqlFragments.of().addSql("*")));

        //from
        fragments.addBlock(FragmentBlock.selectFrom, from(metadata, parameter));

        //where
        where(parameter, metadata)
            .ifPresent(where ->
                           fragments.addBlock(FragmentBlock.where, "where")
                                    .addBlock(FragmentBlock.where, where));

        //order by
        orderBy(parameter, metadata)
            .ifPresent(order -> fragments.addBlock(FragmentBlock.orderBy, "order by")
                                         .addBlock(FragmentBlock.orderBy, order));

        //默认添加allow filtering,否则只能严格按照Cassandra主键查询限制进行查询:只支持主键列及索引列的查询
        //使用非主键列作为查询条件时,性能会降低
        fragments.addBlock(FragmentBlock.after, "allow filtering");

        //分页
        Integer pageSize = parameter.getPageSize();

        if (pageSize != null) {
            //限制最大分页深度
            int maxPageIndex = getQueryMaxPageIndex(parameter);
            if (maxPageIndex > 0) {
                fragments.addBlock(FragmentBlock.paging, PrepareSqlFragments.of("limit ?", maxPageIndex * pageSize));
            }

            //Cassandra不支持跳转分页,只能使用逻辑分页或者滚动分页.
            //从上下文中获取滚动分页ID进行滚动分页
            ByteBuffer pageSate = ScrollPagingUtils
                .getScrollId(parameter.getContext())
                .filter(StringUtils::hasText)
                //解析base64
                .map(str -> ByteBuffer.wrap(Base64.decode(str)))
                .orElse(null);
            return new PagingSqlRequest(fragments.toRequest(), parameter.getPageIndex(), pageSize, pageSate);
        }

        return fragments.toRequest();

    }

    private int getQueryMaxPageIndex(QueryOperatorParameter parameter) {
        Map<String, Object> context = parameter.getContext();
        if (null != context) {
            return CastUtils
                //前端可以传参数到context中来指定分页深度
                .castNumber(context.getOrDefault("maxPageIndex", this.maxPageIndex))
                .intValue();
        }
        return this.maxPageIndex;
    }

    @Override
    protected Optional<SqlFragments> orderBy(QueryOperatorParameter parameter, TableOrViewMetadata metadata) {
        if (CollectionUtils.isEmpty(parameter.getOrderBy())) {
            return Optional.empty();
        }
        //包含了in条件则不能进行排序
        boolean hasIn = parameter
            .getWhere()
            .stream()
            .anyMatch(t -> CastUtils.castArray(t.getValue()).size() > 1);
        if (hasIn) {
            return Optional.empty();
        }
        //主键信息
        CassandraPrimaryKey primaryKey = metadata
            .getFeature(CassandraPrimaryKeyBuilder.ID)
            .flatMap(builder -> builder.builderPrimaryKey(((RDBTableMetadata) metadata)))
            .orElse(null);

        if (primaryKey != null) {
            //条件中包含非主键也不支持排序
            for (Term term : parameter.getWhere()) {
                if (!primaryKey.containsColumn(term.getColumn())) {
                    return Optional.empty();
                }
            }
        }
        PrepareSqlFragments fragments = PrepareSqlFragments.of();
        for (SortOrder sortOrder : parameter.getOrderBy()) {
            //没在主键中定义列不能排序
            if (primaryKey != null && !primaryKey.containsColumn(sortOrder.getColumn())) {
                continue;
            }
            //构造排序
            metadata.getColumn(sortOrder.getColumn())
                    .ifPresent(column -> {
                        if (fragments.isNotEmpty()) {
                            fragments.addSql(",");
                        }
                        fragments.addSql(column.getQuoteName())
                                 .addSql(sortOrder.getOrder().name());
                    });
        }

        return Optional
            .<SqlFragments>of(fragments)
            .filter(SqlFragments::isNotEmpty);
    }
}
